package com.hub.sink;

import com.hub.entity.MyRedisMapper;
import com.hub.entity.MySensor;
import com.hub.entity.SensorReading;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.util.Properties;

/**
 * Redis环境准备：
 * 进入Redis容器
 *      docker exec -it redis-1 bash
 * 连接redis客户端
 *      redis-cli -h 121.41.59.197 -p 6371
 * 查看所有key
 *      hkeys sensor_tempe
 * 查看指定key的值
 *      hget sensor_tempe sensor_1
 * 查看所有key的值
 *      hgetall sensor_tempe
 */
public class SinkFromCustomToRedis {

    public static void main(String[] args) throws Exception {

        // 环境准备
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        // 数据源准备
        // 添加数据源：自定义数据源
        DataStreamSource<SensorReading> dataStreamSource = executionEnvironment.addSource(new MySensor());

        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("121.41.59.197")
                .setPort(6371)
                .build();

        dataStreamSource.addSink( new RedisSink<SensorReading>(config, new MyRedisMapper()) );

        executionEnvironment.execute();

    }

}
